package cn.com.sparkle.paxos.net;

import java.io.Serializable;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import com.google.protobuf.GeneratedMessage;

import cn.com.sparkle.paxos.NodeState;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.AddRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CatchUpRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionPrepareRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionSuccessMessage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionVoteRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.GetNewsSenatorsRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstancePrepareRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceSuccessMessage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceVoteRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.LookUpLatestInstanceIdRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MasterBeatHeartRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SenatorHeatBeatRequest;
import cn.com.sparkle.paxos.net.factory.NetCloseException;
import cn.com.sparkle.paxos.net.factory.PaxosSession;
//import cn.com.sparkle.paxos.net.message.MessagePackage;

public class NetNode {
	private static Logger logger = Logger.getLogger(NetNode.class);

	private PaxosSession session;
	
	private volatile boolean isClose = false;
	private long packageId = 0;
	private ConcurrentHashMap<Long, CallBack> waitFinishMap = new ConcurrentHashMap<Long, CallBack>(
			100000);

	ReentrantLock lock = new ReentrantLock();
	NodeState nodeState;

	public NetNode(PaxosSession session, NodeState nodeState) {
		this.session = session;
		this.nodeState = nodeState;
	}

	public boolean write(GeneratedMessage message, CallBack callBack) {

		MessagePackage.Builder messagePackageBuilder = MessagePackage
				.newBuilder();
		Long id = null;
		if (callBack != null) {
			do{
				id = generatePackageId();
			}while(waitFinishMap.containsKey(id));
			waitFinishMap.put(id, callBack);
			messagePackageBuilder.setId(id).setIsLast(true);
		} else {
			messagePackageBuilder.setId(-1).setIsLast(true);
		}

		// optional SenatorHeatBeatRequest senatorHeartBeatRequest = 3;
		// optional AddRequest addRequest=10;
		// optional ElectionPrepareRequest electionPrepareRequest=12;
		// optional ElectionVoteRequest electionVoteRequest=13;
		// optional LookUpLatestInstanceIdRequest
		// lookUpLatestInstanceIdRequest=16;
		// optional InstancePrepareRequest instancePrepareRequest=17;
		// optional InstanceVoteRequest instanceVoteRequest=18;
		// optional GetNewsSenatorsRequest getNewsSenatorsRequest=19;

		if (message instanceof SenatorHeatBeatRequest) {
			messagePackageBuilder
					.setSenatorHeartBeatRequest((SenatorHeatBeatRequest) message);
		} else if (message instanceof AddRequest) {
			messagePackageBuilder.setAddRequest((AddRequest) message);
		} else if (message instanceof ElectionPrepareRequest) {
			messagePackageBuilder
					.setElectionPrepareRequest((ElectionPrepareRequest) message);
		} else if (message instanceof ElectionVoteRequest) {
			messagePackageBuilder
					.setElectionVoteRequest((ElectionVoteRequest) message);
		} else if (message instanceof ElectionSuccessMessage) {
			messagePackageBuilder
					.setElectionSuccessMessage((ElectionSuccessMessage) message);
		} else if (message instanceof LookUpLatestInstanceIdRequest) {
			messagePackageBuilder
					.setLookUpLatestInstanceIdRequest((LookUpLatestInstanceIdRequest) message);
		} else if (message instanceof InstancePrepareRequest) {
			messagePackageBuilder
					.setInstancePrepareRequest((InstancePrepareRequest) message);
		} else if (message instanceof InstanceVoteRequest) {
			messagePackageBuilder
					.setInstanceVoteRequest((InstanceVoteRequest) message);
		} else if (message instanceof GetNewsSenatorsRequest) {
			messagePackageBuilder
					.setGetNewsSenatorsRequest((GetNewsSenatorsRequest) message);
		} else if (message instanceof MasterBeatHeartRequest) {
			messagePackageBuilder
					.setMasterBeatHeartRequest((MasterBeatHeartRequest) message);
		} else if (message instanceof InstanceSuccessMessage) {
			messagePackageBuilder
					.setInstanceSuccessMessage((InstanceSuccessMessage) message);
		} else if (message instanceof CatchUpRequest) {
			messagePackageBuilder.setCatchUpRequest((CatchUpRequest) message);
		} else {
			logger.error("There is a message can't be processed,please check program.The type of this message is:"
					+ message.getClass());
		}
		MessagePackage mp = messagePackageBuilder.build();
		try {
			session.write(mp);
			logger.debug("add id:" + mp.getId() + " message:"
					+ message.getClass().getName() + " to waitFinish queue");
			return true;
		} catch (NetCloseException e) {
			if(id != null){
				CallBack callback = waitFinishMap.remove(id);
				if (callback != null) {
					callback.fail(this);
				}
			}
			return false;
		}

	}

	public void recieve(MessagePackage o) {
		if (o.getId() == -1)
			return;
		CallBack callBack;
		if (o.getIsLast()) {
			callBack = waitFinishMap.remove(o.getId());
		} else {
			callBack = waitFinishMap.get(o.getId());
		}
		if (callBack != null) {
			callBack.call(this, o);
			logger.debug("recieve id:" + o.getId() + " from "
					+ this.nodeState.getAddress() + " "
					+ callBack.getClass().getName());
		}
	}

	public void close() {
		isClose = true;
		Iterator<CallBack> i = waitFinishMap.values().iterator();
		while (i.hasNext()) {
			CallBack callback = i.next();
			callback.fail(this);
		}
		waitFinishMap.clear();
	}
	public boolean isClose(){
		return isClose;
	}

	private long generatePackageId() {
		packageId = (++packageId) % Long.MAX_VALUE;
		return packageId;
	}

	public PaxosSession getSession() {
		return session;
	}

	public NodeState getNodeState() {
		return nodeState;
	}

}
